package com.gome.ss.flowcrawl.extractor;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.json.JSONArray;
import org.json.JSONObject;

import com.gome.ss.common.AppConfiguer;
import com.gome.ss.common.Utils;

public class FlowCrawlExtractorIndexDataListener implements MessageListener{
	private static Logger logger = Logger.getLogger(FlowCrawlExtractorIndexDataListener.class);
	private static FlowCrawlExtractorIndexDataListener instance = null;
	private static String indexIndices = "";
	private static String indexType = "";
	
	private static Map<String, Object> keepFields = new HashMap<String, Object>();
	
	static {
		keepFields.put("updateStatus", 1);
		keepFields.put("hasStorage", 0);
		keepFields.put("tag", 0);
		keepFields.put("storageDate", "1970-01-01 00:00:00:000");
	}
	
	private ConnectionFactory connectionFactory = null;
	private Connection connection = null;
	private Session session = null;
	private MessageConsumer consumer = null;
	private Client esClient = ElasticsearchClientFactory.getClient();
	
	private FlowCrawlExtractorIndexDataListener() {
		indexIndices = AppConfiguer.getStringByKey("ESIndexIndices");
		indexType = AppConfiguer.getStringByKey("ESIndexType");
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                AppConfiguer.getStringByKey("activeMQHost"));
        
        try {
			connection = connectionFactory.createConnection();
			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue(FlowCrawlExtractorManager.esdqName);
			consumer = session.createConsumer(destination);
			consumer.setMessageListener(this);
		} catch (Exception e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	public static void createListener() {
		if (instance == null) {
			instance = new FlowCrawlExtractorIndexDataListener();
		}
	}
	
	public static FlowCrawlExtractorIndexDataListener getInstance() {
		if (instance == null) {
			instance = new FlowCrawlExtractorIndexDataListener();
		}
		return instance;
	}
	
	private void fixData(String id, JSONObject obj) {
		
		obj.remove("parentPage");
		obj.remove("_id");
		
		QueryBuilder query = QueryBuilders.termQuery("id", id);
		
		SearchResponse sresponse = esClient.prepareSearch(indexIndices)
		  .setTypes(indexType)
		  .setQuery(query)
		  .execute().actionGet();
		
		if (sresponse.getHits().totalHits() > 0) {
			String source = sresponse.getHits().hits()[0].getSourceAsString();
			JSONObject sobj = new JSONObject(source);
			Set<String> keepFieldKeys = keepFields.keySet();
			
			for (String key : keepFieldKeys) {

				if (sobj.has(key)) {
					if (key.equals("updateStatus")) {
						if (obj.has("date") && sobj.has("date")) {
							String dateString = obj.getString("date");
							String sdateString = sobj.getString("date");
							
							try {
								SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
								Date date = sdf.parse(dateString);
								Date sdate = sdf.parse(sdateString);
								if (sdate.compareTo(date) < 0)
									obj.put(key, 1);
								else
									obj.put(key, sobj.get(key));
							} catch (ParseException e) {
								logger.error(Utils.getTrace(e));
								obj.put(key, sobj.get(key));
							}
						} else {
							obj.put(key, keepFields.get(key));
						}
						continue;
					}
					obj.put(key, sobj.get(key));
				} else
					obj.put(key, keepFields.get(key));
			}
			
			DeleteResponse response = esClient.prepareDelete(indexIndices, indexType, id).execute().actionGet();
			if (response.isFound())
				logger.info("Delete index " + response.getId());
			
		} else {
			Set<String> keepFieldKeys = keepFields.keySet();
			for (String key : keepFieldKeys) {
				obj.put(key, keepFields.get(key));
			}
		}
	}
	
	private void doIndex(String id, JSONObject obj) {
		
		JSONArray keyWord = obj.getJSONArray("keyWord");
		String keyWordWhiteSpace = "";
		
		for (int i = 0; i < keyWord.length(); i ++) {
			keyWordWhiteSpace += keyWord.getString(i);
			if (i < keyWord.length()-1)
				keyWordWhiteSpace += " ";
		}
		
		obj.remove("keyWord");
		obj.put("keyWord", keyWordWhiteSpace);
		
		fixData(id, obj);
		
		IndexResponse response = esClient.prepareIndex(indexIndices, indexType)
					.setSource(obj.toString()).setId(id).execute().actionGet();
		
		logger.info("Do index " + response.getId());
	}
	
	@Override
	public void onMessage(Message message) {
		// TODO Auto-generated method stub
		TextMessage textMess = (TextMessage)message;
		try {
			String text = textMess.getText();
			JSONObject obj = new JSONObject(text);
			
			String id = obj.getString("id");
			
			doIndex(id, obj);
			
		} catch (JMSException e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	public void appendNotAnalyzedMappingField(String field) throws Exception {
		ElasticsearchClientFactory.appendNotAnalyzedMappingField(esClient, indexIndices, indexType, field);
	}
	
	public static void main(String[] args) throws Exception {
		Client esClient = ElasticsearchClientFactory.getClient();
		ElasticsearchClientFactory.createMapping(esClient, "flowcrawl_extractor_index_v1", "flowcrawl_extractor_data");
		ElasticsearchClientFactory.appendNotAnalyzedMappingField(esClient, "flowcrawl_extractor_index_v1",
				"flowcrawl_extractor_data", "weixinname");
	}
}
